import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.junit.Test;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.util.Properties;
import java.util.concurrent.ExecutionException;

import static java.lang.System.in;

/**********************************
 * @author wmx
 * @date 2019/11/27 15:30
 * @description:
 * @version 1.0
 *
 * 问题一
 * java.io.IOException: Can't resolve address: node1:9092
 * java.io.IOException: Can't resolve address: node2:9092
 * java.io.IOException: Can't resolve address: node3:9092
 * 当Kafka broker启动时，它会在ZK上注册自己的IP和端口号，客户端就通过这个IP和端口号来连接。
 * 在AWS这种IaaS环境下，由于java.net.InetAddress.getCanonicalHostName调用拿到的HostName是主机名，所以默认注册到ZK上的是主机名
 * 需要配置server.properties
 * advertised.listeners=PLAINTEXT://192.168.16.11:9092

 *
 ************************************/
public class ProducerKafka {
    public static void producerData() throws IOException {
        Properties props = new Properties();
        //kafka 集群，broker-list

        props.put("bootstrap.servers", "node1:9092");

        props.put("acks", "all");

        //重试次数

        props.put("retries", 1);

        //批次大小

        props.put("batch.size", 16384);

        //等待时间

        props.put("linger.ms", 1);

        //RecordAccumulator 缓冲区大小 props.put("buffer.memory", 33554432);

        props.put("key.serializer",
                "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer",
                "org.apache.kafka.common.serialization.StringSerializer");

        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(props);
        while (true) {
            System.out.println("请输入要发送的信息：");
            InputStream in = System.in;
            BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(in));String str = bufferedReader.readLine();
            if (str != null) {
                try {
                    kafkaProducer.send(new ProducerRecord<String, String>("test8",0,"2" ,str)).get();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } catch (ExecutionException e) {
                    e.printStackTrace();
                }
                if (str.equals("end")){
                    kafkaProducer.close();
                }
            }
            System.out.println("你输入的是："+str);
        }


    }

    public static void main(String[] args) throws Exception{
        producerData();
    }
}
